package com.jch.lx;

import org.apache.flink.api.common.accumulators.IntCounter;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSink;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.MapOperator;
import org.apache.flink.configuration.Configuration;

public class Count {

    public static void main(String[] args) {

        //创建环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        
        //读取数据
        DataSource<String> source = env.readTextFile("");

        MapOperator<String, String> result = source.map(new RichMapFunction<String, String>() {

            //创建累加器
            private IntCounter elementCounter = new IntCounter();
            Integer count = 0;

            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                //注册累加器
                getIterationRuntimeContext().addAccumulator("elementCounter", elementCounter);
            }

            @Override
            public String map(String s) throws Exception {

                //使用累加器
                this.elementCounter.add(1);
                count += 1;
                System.out.println("不使用累加器统计的结果" + count);
                return s;
            }
        }).setParallelism(2);


        //sink
        DataSink<String> sink = result.writeAsText("");





    }



}
